/*
 * net_s.cpp
 *
 *  Created on: Sep 27, 2013
 *      Author: killerg
 */




#include "net_s.h"
#include "util.h"

#include <event2/buffer.h>
#include <unistd.h>
#include <string.h>
#include <sys/socket.h>
#include <stdio.h>
#include <assert.h>

bool net_s::init(int max_client, code_cb_t code_cb, size_t max_msg_size, bool nodelay, int deal_per_tick) {
    this->code_cb = code_cb;
    this->bytes_recv = 0;
    this->bytes_send = 0;
    this->max_fdnum = max_client + 1;
    this->cur_fdnum = 0;
    this->accept_fd = -1;
    this->accept_tr = NULL;
    this->max_msg_size = max_msg_size;
    this->nodelay = nodelay;
    this->deal_per_tick = deal_per_tick;

    // 创建epollfd
    epfd = epoll_create(this->max_fdnum);
    if (epfd < 0) {
        UTIL_LOG(E, "net_s net_s::init epoll_create: %s", strerror(errno));
        return false;
    }

    events = new epoll_event[this->max_fdnum];
    if (events == NULL) {
        UTIL_LOG(E, "net_s net_s::init new epoll_event: %s", strerror(errno));
        return false;
    }

    return true;
}
void net_s::destroy() {
    if (events != NULL) {
        delete[] events;
        events = NULL;
    }

    // 销毁epollfd
    if (epfd >= 0) {
        close(epfd);
        epfd = -1;
    }
}

void net_s::update() {
    // epoll wait
    int num = epoll_wait(epfd, events, max_fdnum, 0);
    if (num < 0) {
        UTIL_LOG(E, "net_s %s_%d err: %s", __FUNCTION__, __LINE__, strerror(errno));
        assert(0);
        return;
    }

    // for all events
    // #pragma omp parallel for
    for (int i=0; i<num; ++i) {
        epoll_event* ev = events+i;
        tran_s* tr = (tran_s*)ev->data.ptr;

        // if is listen
        if (accept_tr == tr) {
            if (ev->events & EPOLLIN) {
                if (cur_fdnum < max_fdnum) {
                    do {
                        int newfd = accept(tr->fd, NULL, NULL);
                        if (newfd < 0) {
                            if (errno != EWOULDBLOCK && errno != EAGAIN) {
                                UTIL_LOG(E, "net_s %s_%d err: %s", __FUNCTION__, __LINE__, strerror(errno));
                            }
                            break;
                        }

                        auto tr = new tran_s(newfd, this);
                        if (acceptcb)
                            acceptcb(tr);

                        accept_queue.push(tr);
                        ++cur_fdnum;
                        if (cur_fdnum >= max_fdnum)
                            break;
                    } while (true);
                }
            }
        // normal fd
        } else {
            // be closed
            if (tr->closed)
                continue;

            bool need_close = false;
            if (ev->events & (EPOLLERR|EPOLLHUP)) {
                need_close = true;
            } else {
                if (ev->events & EPOLLIN) {
                    int res = evbuffer_read(tr->input, tr->fd, -1);
                    if (res == 0) {
                        need_close = true;
                    } else if (res < 0) {
                        if (errno != EWOULDBLOCK && errno != EAGAIN) {
                        	UTIL_LOG(E, "net_s %s_%d err: %d_%s", __FUNCTION__, __LINE__, errno, strerror(errno));
                            need_close = true;
                        }
                    }

                    bytes_recv += res;
                }

                if (!need_close && (ev->events & EPOLLOUT)) {
                    int res = evbuffer_write(tr->output, tr->fd);
                    if (res < 0) {
                        if (errno != EWOULDBLOCK && errno != EAGAIN) {
                            UTIL_LOG(E, "net_s %s_%d err: %d_%s", __FUNCTION__, __LINE__, errno, strerror(errno));
                            need_close = true;
                        }
                    } else {
                        bytes_send += res;
                    }

                    int need_write = tr->get_output_len();
                    if (need_write <= 0)
                        tr->deal_epoll(EPOLL_CTL_MOD, EPOLLIN/*|EPOLLET*/);
                }
            }

            if (need_close) {
                tr->disconnect(true);
            }
        }
    }

    // deal queued accept
    for (int i=0; i<deal_per_tick && !accept_queue.empty(); ++i) {
        auto tr = accept_queue.front();
        accept_queue.pop();

        if (nodelay) {
            int yes = 1;
            setsockopt(tr->fd, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(int));
        }
        evutil_make_socket_nonblocking(tr->fd);
        if (!tr->deal_epoll(EPOLL_CTL_ADD, EPOLLIN/*|EPOLLET*/)) {
            tr->disconnect(false);
            continue;
        }
        logoncb(tr);
    }

    // deal queued close
    for (int i=0; i<deal_per_tick && !gc_queue.empty(); ++i) {
        auto tr = gc_queue.front();
        gc_queue.pop();

        if (tr->need_logoff)
            logoffcb(tr);

        close(tr->fd);
        --cur_fdnum;
        delete tr;
    }
}

bool net_s::start_accept(const char* ip, int port, handler_t logoncb, handler_t logoffcb, handler_t acceptcb) {
    if (cur_fdnum >= max_fdnum)
        return false;

    this->logoncb = logoncb;
    this->logoffcb = logoffcb;
    this->acceptcb = acceptcb;

    // 建立listenfd
    int listenfd = socket(AF_INET, SOCK_STREAM, 0);
    if (listenfd < 0) {
        UTIL_LOG(E, "net_s %s_%d err: %s", __FUNCTION__, __LINE__, strerror(errno));
        return false;
    }
    evutil_make_socket_nonblocking(listenfd);
    evutil_make_listen_socket_reuseable(listenfd);

    sockaddr addr;
    int addr_len = sizeof(addr);
    if (evutil_parse_sockaddr_port(fmt_cstr("%s:%d", ip, port), &addr, &addr_len) < 0) {
        UTIL_LOG(E, "net_s %s_%d err: %s", __FUNCTION__, __LINE__, strerror(errno));
        return false;
    }

    if (bind(listenfd, &addr, addr_len) < 0) {
        UTIL_LOG(E, "net_s %s_%d err: %s", __FUNCTION__, __LINE__, strerror(errno));
        return false;
    }

    if (listen(listenfd, 1000) < 0) {
        UTIL_LOG(E, "net_s %s_%d err: %s", __FUNCTION__, __LINE__, strerror(errno));
        return false;
    }


    auto tr = new tran_s(listenfd, this);
    if (NULL == tr) {
        UTIL_LOG(E, "net_s %s_%d err: %s", __FUNCTION__, __LINE__, strerror(errno));
        close(listenfd);
        listenfd = -1;
        return false;
    }

    if (!tr->deal_epoll(EPOLL_CTL_ADD, EPOLLIN/*|EPOLLET*/)) {
        close(listenfd);
        listenfd = -1;
        safe_del(tr);
        return false;
    }

    ++cur_fdnum;
    accept_fd = listenfd;
    accept_tr = tr;

    return true;
}

void net_s::stop_accept() {
    if (accept_tr) {
        accept_tr->disconnect(false);
        delete accept_tr;
    }
    accept_tr = NULL;
    accept_fd = -1;
}

int tran_s::get_output_len() {
    return evbuffer_get_length(output);
}

void tran_s::disconnect(bool need_logoff) {
    if (closed)
        return;

    this->need_logoff = need_logoff;
    closed = true;
    ns->gc_queue.push(this);
}

bool tran_s::deal_epoll(int ctrl, int events) {
    epoll_event ev;
    ev.events = events;
    ev.data.ptr = this;

    if (epoll_ctl(ns->epfd, ctrl, fd, &ev) < 0) {
        UTIL_LOG(E, "net_s %s_%d err: %s", __FUNCTION__, __LINE__, strerror(errno));
        return false;
    }

    return true;
}

tran_s::tran_s(int fd, net_s* ns) {
    this->bytes_recv = 0;
    this->bytes_send = 0;
    this->need_logoff = false;
    this->closed = false;
    this->fd = fd;
    this->ns = ns;
    this->recv_index = INIT_MSG_INDEX;
    input = evbuffer_new();
    output = evbuffer_new();
}

tran_s::~tran_s() {
    need_logoff = false;
    closed = false;
    fd = -1;
    ns = NULL;
    if (NULL != input)
        evbuffer_free(input);

    if (NULL != output)
        evbuffer_free(output);

    input = output = NULL;
}

